package com.feidee.fdhadoop.hbase;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdhadoop.kerberos.KerberosAuthentication;
import com.feidee.fdhadoop.logCollect.LogCollectService;
import com.feidee.fdhadoop.constant.Constant;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import com.feidee.fdhadoop.kerberos.KerberosAuthentication;

import java.io.IOException;
import java.util.*;

public class HBaseManager {
	private static Logger logger = Logger.getLogger(HBaseManager.class);
	private static Configuration conf = null;
	private Table htable = null;
	private ResultScanner resultScanner = null;
	private static Connection connection = null;
	static{
		try {
			KerberosAuthentication.authenticate();
			conf = HBaseConfiguration.create();
			conf.set("hbase.zookeeper.property.clientPort", CustomConfiguration.getString("hbase.zookeeper.property.clientPort","2181"));
			conf.set("hbase.zookeeper.quorum", CustomConfiguration.getString("hbase.zookeeper.quorum","10.200.1.11,10.200.1.19,10.200.1.14,10.200.1.29,10.200.1.34"));
			conf.set("hbase.client.scanner.caching", CustomConfiguration.getString("hbase.client.scanner.caching","100"));
			conf.set("hbase.rpc.timeout", CustomConfiguration.getString("hbase.rpc.timeout","6000"));//rpc的超时时间
			conf.set("ipc.socket.timeout", CustomConfiguration.getString("ipc.socket.timeout","2000"));//socket建立链接的超时时间，应该小于或者等于rpc的超时时间
			conf.set("hbase.client.retries.number", CustomConfiguration.getString("hbase.client.retries.number","3"));//重试次数
			conf.set("hbase.client.pause", CustomConfiguration.getString("hbase.client.pause","100"));//重试休眠时间，默认为1s
			conf.set("zookeeper.recovery.retry", CustomConfiguration.getString("zookeeper.recovery.retry","3"));//zk重试次数
			connection = ConnectionFactory.createConnection(conf);
		} catch (IOException e) {
			logger.error("HBase Connection init failed ",e);
		}
	}

	public HBaseManager(String tableName) throws IOException {
		if(connection != null){
			htable = connection.getTable(TableName.valueOf(tableName));
		}
	}

	public HBaseManager(){
	}

	public ResultScanner getResultScanner() throws IOException {
		Scan scan = new Scan();
		scan.setCacheBlocks(false);
		if(htable != null){
			resultScanner = htable.getScanner(scan);
		}else{
			resultScanner = null;
		}
		return resultScanner;
	}

	public ResultScanner getResultScanner(FilterList filter) throws IOException {
		Scan scan = new Scan();
		scan.setCacheBlocks(false);
		scan.setFilter(filter);
		if(htable != null){
			resultScanner = htable.getScanner(scan);
		}else{
			resultScanner = null;
		}
		return resultScanner;
	}

	public ResultScanner getResultScanner(Scan scan) throws IOException {
		resultScanner = htable.getScanner(scan);
		return resultScanner;
	}

	public Result get(String rowKey) throws IOException {
		Get get = new Get(Bytes.toBytes(rowKey));
		Result rs = null;
		rs = htable.get(get);
		return rs;
	}

	public Result get(String rowKey,Filter filter) throws IOException {
		Get get = new Get(Bytes.toBytes(rowKey));
		get.setFilter(filter);
		Result rs = null;
		rs = htable.get(get);
		return rs;
	}

	public static Result get(String table, String rowKey, Filter filter) throws IOException {
		Table htable = null;
		Result rs = null;
		try {
			if(connection != null){
				htable = connection.getTable(TableName.valueOf(table));
			}
			if(htable != null && rowKey != null){
				Get get = new Get(Bytes.toBytes(rowKey));
				get.setFilter(filter);
				rs = htable.get(get);
			}
		} finally {
			try {
				if(htable != null)
					htable.close();
			} catch (IOException e) {
			}
		}
		return rs;
	}

	public static Result[] get(String table, List<String> rowKeys) throws IOException {
		Table htable = null;
		try {
			if(connection != null){
				htable = connection.getTable(TableName.valueOf(table));
			}
			if(htable != null && rowKeys != null && rowKeys.size() > 0){
				List<Get> gets = new ArrayList<Get>();
				for (String rowKey : rowKeys) {
					Get get = new Get(Bytes.toBytes(rowKey));
					gets.add(get);
				}
				return htable.get(gets);
			}
		} finally {
			try {
				if(htable != null)
					htable.close();
			} catch (IOException e) {
			}
		}
		return null;
	}

	public static Result[] get(String table, List<String> rowKeys, Filter filter) throws IOException {
		Table htable = null;
		try {
			if(connection != null){
				htable = connection.getTable(TableName.valueOf(table));
			}
			if(htable != null && rowKeys != null && rowKeys.size() > 0){
				List<Get> gets = new ArrayList<Get>();
				for (String rowKey : rowKeys) {
					Get get = new Get(Bytes.toBytes(rowKey));
					get.setFilter(filter);
					gets.add(get);
				}
				return htable.get(gets);
			}
		}finally {
			try {
				if(htable != null)
					htable.close();
			} catch (IOException e) {
			}
		}
		return null;
	}

	public void closeHtable(){
		if (resultScanner != null) {
			resultScanner.close();
		}
		try {
			if (htable != null) {
				htable.close();
			}
		} catch (IOException e) {
		}
	}

	public static void closeConnection(){
		try {
			connection.close();
		} catch (IOException e) {
		}
	}

	public static Table getHTable(String tabel) throws IOException {
		Table table = null;
		if(connection != null)
			table = connection.getTable(TableName.valueOf(tabel));
		return table;
	}

	public static boolean tableExists(String table) throws IOException {
		Admin admin = null;
		boolean exist = false;
		try{
			admin = connection.getAdmin();
			exist = admin.tableExists(TableName.valueOf(table));
		}finally {
			if(admin != null){
				try {
					admin.close();
				} catch (IOException e) {
				}
			}
		}
		return exist;
	}

	public static void createTable(String table, String ... familys) throws IOException {
		Admin admin = null;
		try{
			admin = connection.getAdmin();
			TableName tableName = TableName.valueOf(table);
			if(!admin.tableExists(tableName)){
				HTableDescriptor tdesc = new HTableDescriptor(tableName);
				for(String family : familys){
					HColumnDescriptor cdesc = new HColumnDescriptor(family);
					tdesc.addFamily(cdesc);
				}
				admin.createTable(tdesc);
			}
		}finally {
			if(admin != null){
				try {
					admin.close();
				} catch (IOException e) {
				}
			}
		}

	}

	public static void addCells(String table,String rowkey,String family,Map<String, String> records) throws IOException {
		Table htable = null;
		try{
			htable = getHTable(table);
			Put put = new Put(Bytes.toBytes(rowkey));
			Iterator<String> iterator = records.keySet().iterator();
			while(iterator.hasNext()){
				String key = iterator.next();
				String value = records.get(key);
				put.addColumn(Bytes.toBytes(family), Bytes.toBytes(key), Bytes.toBytes(value));
				htable.put(put);
			}

		}finally {
			if(htable != null){
				try {
					htable.close();
				} catch (IOException e) {
				}
			}
		}
	}

	public static void addRecordsBatch(String table, String family,List<Map<String, String>> list) throws IOException {
		BufferedMutator mutator = null;
		try{
			mutator = connection.getBufferedMutator(TableName.valueOf(table));
			ArrayList<Put> puts = new ArrayList<Put>();
			for(Map<String, String> records : list){
				String rowkey = records.get("rowkey");
				if(rowkey != null){
					Put put = new Put(Bytes.toBytes(rowkey));
					Iterator<String> iterator = records.keySet().iterator();
					while(iterator.hasNext()){
						String key = iterator.next();
						if(!key.equals("rowkey")){
							String value = records.get(key);
							put.addColumn(Bytes.toBytes(family), Bytes.toBytes(key), Bytes.toBytes(value));
						}
					}
					puts.add(put);
					if(puts.size() >= CustomConfiguration.getInt("hbase.batch.save.size",1000)){
						mutator.mutate(puts);
						mutator.flush();
						puts.clear();
					}
				}
			}
			if(puts.size() >= 0){
				mutator.mutate(puts);
				mutator.flush();
				puts.clear();
			}
		}finally {
			if(mutator != null){
				try {
					mutator.close();
				} catch (IOException e) {
				}
			}
		}
	}

	public static void deleteColumnBatch(String table, List<String> rowList, String family) throws IOException {
		BufferedMutator mutator = null;
		try {
			mutator = connection.getBufferedMutator(TableName.valueOf(table));
			ArrayList<Delete> delList = new ArrayList<Delete>();
			for (String rowKey : rowList) {
				Delete delete = new Delete(Bytes.toBytes(rowKey));
				delete.addFamily(Bytes.toBytes(family));
				delList.add(delete);
				if (delList.size() > CustomConfiguration.getInt("hbase.batch.save.size",1000)) {
					mutator.mutate(delList);
					mutator.flush();
					delList.clear();
				}
			}

			if (delList.size() > 0) {
				mutator.mutate(delList);
				mutator.flush();
				delList.clear();
			}
		} finally {
			try {
				mutator.close();
			} catch (IOException e) {
			}
		}
	}

	public static void deleteColumnBatch(String table, List<String> rowList, String family, String qualifier) throws IOException {
		BufferedMutator mutator = null;
		try {
			mutator = connection.getBufferedMutator(TableName.valueOf(table));
			ArrayList<Delete> delList = new ArrayList<Delete>();
			for (String rowKey : rowList) {
				Delete delete = new Delete(Bytes.toBytes(rowKey));
				delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(qualifier));
				delList.add(delete);
				if (delList.size() > CustomConfiguration.getInt("hbase.batch.save.size",1000)) {
					mutator.mutate(delList);
					mutator.flush();
					delList.clear();
				}
			}

			if (delList.size() > 0) {
				mutator.mutate(delList);
				mutator.flush();
				delList.clear();
			}
		}finally {
			try {
				mutator.close();
			} catch (IOException e) {
			}
		}
	}

	public static void deleteColumnBatch(String table,String columnFamily, List<List<String>> list) throws IOException {
		BufferedMutator mutator = null;
		try {
			mutator = connection.getBufferedMutator(TableName.valueOf(table));
			ArrayList<Delete> delList = new ArrayList<Delete>();
			for(List<String> records : list){
				String rowkey = records.get(0);
				if(rowkey != null){
					Delete delete = new Delete(Bytes.toBytes(rowkey));
					for(int i = 1; i < records.size(); i++){
						delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(records.get(i)));
					}
					delList.add(delete);
				}
				if (delList.size() > CustomConfiguration.getInt("hbase.batch.save.size",1000)) {
					mutator.mutate(delList);
					mutator.flush();
					delList.clear();
				}
			}
			if (delList.size() > 0) {
				mutator.mutate(delList);
				mutator.flush();
				delList.clear();
			}
		}finally {
			try {
				mutator.close();
			} catch (IOException e) {
			}
		}
	}
}




